package com.atguigu.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.util.DimUtil;
import com.atguigu.util.DruidPhoenixDSUtil;
import com.atguigu.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.SQLException;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;


public abstract class MyAsyncJoinFunction<T> extends RichAsyncFunction<T, T> implements MyJoinFunction<T> {
    DruidDataSource dataSource = null;
    ThreadPoolExecutor threadPoolExecutor = null;
    String tableName;

    public MyAsyncJoinFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        dataSource = DruidPhoenixDSUtil.getDataSource();
        threadPoolExecutor = ThreadPoolUtil.getInstance();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                DruidPooledConnection conn = null;
                try {
                    conn = dataSource.getConnection();
                    JSONObject dimInfo = DimUtil.getDimInfo(conn, tableName, getKey(input));
                    join(input,dimInfo);
                }catch (Exception e){
                    e.printStackTrace();
                }finally {
                    if (conn != null){
                        try {
                            conn.close();
                        } catch (SQLException throwables) {
                            throwables.printStackTrace();
                        }
                    }
                }
                resultFuture.complete(Collections.singleton(input));
            }
        });
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        resultFuture.complete(Collections.singleton(input));
    }
}
